package com.baishancloud.log.live.spider

import org.apache.flink.api.common.state.{MapState, MapStateDescriptor, StateTtlConfig}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.util

class SpiderWindowAgg(parameterTool: ParameterTool) extends ProcessWindowFunction[SpiderLog, SpiderResult, SpiderLogKey, TimeWindow] {

  //状态未被修改最长保留时间
  var ttlHour: Long = _
  var uuidState: MapState[String, String] = _

  /**
   * 初始化list状态，设置状态ttl
   */
  override def open(parameters: Configuration): Unit = {
    ttlHour = parameterTool.getLong(allowOldestHour, 6)
    val mapStateDescriptor: MapStateDescriptor[String, String] = new MapStateDescriptor[String, String]("所有已出现过的uuid", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)
    val ttlConfig: StateTtlConfig = StateTtlConfig
      .newBuilder(Time.hours(ttlHour))
      .updateTtlOnCreateAndWrite()
      .returnExpiredIfNotCleanedUp()
      .useProcessingTime()
      .updateTtlOnCreateAndWrite()
      .build()
    mapStateDescriptor.enableTimeToLive(ttlConfig)
    uuidState = getRuntimeContext.getMapState(mapStateDescriptor)
  }

  /**
   * 只处理窗口内时间最小的那些日志对象的数据，将他们的clientId存储到
   */
  override def process(key: SpiderLogKey, context: Context, elements: Iterable[SpiderLog], out: Collector[SpiderResult]): Unit = {
    var uv: Long = 0
    var flag: Boolean = false
    elements.foreach(element => {
      if (!uuidState.contains(element.uuid)) {
        //表示有新的uuid出现，需要输出新的uv统计结果
        flag = true
        uuidState.put(element.uuid, "")
      }
    })
    if (flag) {
      val keysIter: util.Iterator[String] = uuidState.keys().iterator()
      while (keysIter.hasNext) {
        keysIter.next()
        uv += 1
      }
      out.collect(SpiderResult(key.time_at_1min, key.domain, key.country, key.province, key.isp, key.log_type, uv))
    }
  }
}
